package cn.kgc.gmall.app.func;

import cn.kgc.gmall.bean.TableProcess;
import cn.kgc.gmall.common.GmallConfig;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {

    // 定义侧输出流
    private OutputTag<JSONObject> outputTag;
    // 定义状态变量
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;

    private Connection conn;

    public TableProcessFunction(OutputTag<JSONObject> outputTag, MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.outputTag = outputTag;
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        //初始化Phoenix连接
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        conn = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);

    }

    /**
     * 普通流
     * @param value 流中的元素
     * @param ctx 上下文参数
     * @param out 向下游写出
     * @throws Exception
     */
    @Override
    public void processElement(JSONObject value, BroadcastProcessFunction<JSONObject, String, JSONObject>.ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        // 在主流里面将数据输出到不通过的输出流
        // {"database":"gmall_2022","table":"activity_info","type":"delete","ts":1661916124,"data":{"id":1564815757992218659,"user_id":1200,}}
        // 封装好key
        String table = value.getString("table");
        String type = value.getString("type");
        JSONObject data = value.getJSONObject("data");
        if(type.equals("bootstrap-insert")){
            type = "insert";
        }
        String key = table + ":" + type;

        // 从上下文中获取状态变量
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        TableProcess tableProcess = broadcastState.get(key);
        if (tableProcess == null ){
            // 没有匹配到的数据废弃掉
            //System.out.println(value.toJSONString());
        }else{
            value.put("sink_table",tableProcess.getSinkTable());
            // 如果配置表中的sinkType正好是hbase 就往侧输出流输出
            if (tableProcess.getSinkType().equals(TableProcess.SINK_TYPE_HBASE)){
                // 过滤掉不要的属性
                filterColumn(data,tableProcess.getSinkColumns());
                ctx.output(outputTag,value);
            }else{
                // 如果配置表中的sinkType是kafka往正常流中输出
                out.collect(value);
            }
        }

    }

    /**
     * 广播流
     * @param jsonStr  每个元素
     * @param ctx 上下文参数
     * @param out 向下游写出
     * @throws Exception
     */
    @Override
    public void processBroadcastElement(String jsonStr, BroadcastProcessFunction<JSONObject, String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
        // 在广播流中将数据存入到状态变量中
        // 先转成json对象

        JSONObject jsonObject = JSON.parseObject(jsonStr);
        // 获取json中的数据
        String data = jsonObject.getString("data");
        // 将字段数据 转化为实体类
        TableProcess tableProcess = JSON.parseObject(data, TableProcess.class);

        // 取出里面的数据
        // 操作类型 insert update  当读取到的是历史数据的时候就会编程 bootstrap-insert
        String operateType = tableProcess.getOperateType();
        if (operateType.equals("bootstrap-insert")){
            // 数据修复
            operateType = "insert";
        }

        // 源表
        String sourceTable = tableProcess.getSourceTable();
        // 目标表
        String sinkTable = tableProcess.getSinkTable();
        // 目标数据下游类型  hbase  kafka
        String sinkType = tableProcess.getSinkType();
        // 当插入数据的时候 id
        String sinkPk = tableProcess.getSinkPk();
        // 往下游写的时候需要那些字段
        String sinkColumns = tableProcess.getSinkColumns();
        // 扩展字段
        String sinkExtend = tableProcess.getSinkExtend();

        // TODO 在往hbase写的时候 先建数据库 gmall_2022_realtime 建表
        if (TableProcess.SINK_TYPE_HBASE.equals(sinkType) && "insert".equals(operateType)){
            createTable(sinkTable,sinkColumns,sinkPk,sinkExtend);
        }

        // 拼接key
        String key = sourceTable + ":" + operateType;

        // 使用状态变量描述
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        broadcastState.put(key,tableProcess);
    }
    /**
     * 读取到配置表中的信息在phoenix建表
     */
    private void createTable(String sinkTable, String sinkColumns, String sinkPk,String sinkExtend) {

        // 如果pk为null 就设置为id
        if ("".equals(sinkPk)){
            sinkPk = "id";
        }
        // id,activity_name,activity_type,activity_desc,start_time,end_time,create_time
        String[] columns = sinkColumns.split(",");
        // 开始建表
        StringBuffer sql = new StringBuffer(" create table if not exists "+ GmallConfig.HBASE_SCHEMA+"."+sinkTable+" ( ");
        // create table if not existe GMALL_2022_REALTIME.user_info(id varchar primary key ,属性名 varchar,... )
        for (int i = 0; i < columns.length; i++) {
            if (columns[i].equals(sinkPk)){
                sql.append(columns[i]+" VARCHAR PRIMARY KEY ");
            }else{
                sql.append(columns[i]+" VARCHAR ");
            }
            //判断是否添加逗号
            if (i < columns.length-1){
                sql.append(" , ");
            }
        }

        sql.append(" ) "+sinkExtend);
        //System.out.println(sql);
        PreparedStatement ps = null;
        try {
            // 创建语句执行者
            ps = conn.prepareStatement(sql.toString());
            // 执行
            ps.execute();
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            if (ps != null){
                try {
                    ps.close();
                } catch (SQLException e) {
                    throw new RuntimeException(e);
                }
            }
        }


    }




    // TODO 过滤每条数据判断是否包含要保留的字段
    //对dataJsonObj中的属性进行过滤
    //dataJsonObj："data":{"id":12,"tm_name":"atkgc","logo_url":"/static/beijing.jpg"}
    //sinkColumns : id,tm_name
    private void filterColumn(JSONObject dataJsonObj, String sinkColumns){
        String[] fieldArr = sinkColumns.split(",");
        // 为了判断是否包含转化为列表
        List<String> fieldList = Arrays.asList(fieldArr); // 将数组转化为列表
        Set<Map.Entry<String, Object>> entries = dataJsonObj.entrySet();
        // 函数编程
        //fieldList.removeIf(x -> !x.contains(entries.iterator().next().getKey()));
        entries.removeIf(x -> !fieldList.contains(x.getKey()));
    }


    /*private void filterColumn(JSONObject dataJsonObj, String sinkColumns) {
        String[] fieldArr = sinkColumns.split(",");
        //将数据转换为List集合，方便后面通过判断是否包含key
        List<String> fieldList = Arrays.asList(fieldArr);

        //获取json对象的封装的键值对集合
        Set<Map.Entry<String, Object>> entrySet = dataJsonObj.entrySet();
        //获取迭代器对象   因为对集合进行遍历的时候，需要使用迭代器进行删除
        Iterator<Map.Entry<String, Object>> it = entrySet.iterator();
        //对集合中元素进行迭代
        for (;it.hasNext();) {
            //得到json中的一个键值对
            Map.Entry<String, Object> entry = it.next();
            //如果sink_columns中不包含  遍历出的属性    将其删除
            if(!fieldList.contains(entry.getKey())){
                it.remove();
            }
        }
    }*/


}
